package com.kqkj.dyrule.marketing.functions;

import com.alibaba.fastjson.JSON;
import com.kqkj.dyrule.marketing.beans.*;
import com.kqkj.dyrule.marketing.utils.StateDescContainer;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class DynamicKeyByReplicateFunctionV2 extends BroadcastProcessFunction<EventBean, MaxwellLogBean, DynamicKeyedBean> {

    BroadcastState<String, List<RuleStateBean>> broadcastState;

    /**
     * 动态keyby数据复制
     * @param eventBean
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(EventBean eventBean, ReadOnlyContext ctx, Collector<DynamicKeyedBean> out) throws Exception {

        for (Map.Entry<String, List<RuleStateBean>> entry : broadcastState.entries()) {
            String keyByField = entry.getKey();
            StringBuilder sb = new StringBuilder();
            String[] fieldNames = keyByField.split(",");
            //拼装keyByField中指定每一个字段的值
            for (String fieldName : fieldNames) {
                Class<?> beanClass = Class.forName("com.kqkj.dyrule.marketing.beans.EventBean");
                Field declaredField = beanClass.getDeclaredField(fieldName);
                declaredField.setAccessible(true);
                String fieldValue = (String)declaredField.get(eventBean);
                sb.append(fieldValue).append(",");
            }
            String keyByValue = sb.toString().substring(0, sb.length() - 1);
            eventBean.setKeyByValue(keyByValue);
            DynamicKeyedBean dynamicKeyedBean = new DynamicKeyedBean(keyByValue, keyByField, eventBean);
            out.collect(dynamicKeyedBean);
        }

    }

    @Override
    public void processBroadcastElement(MaxwellLogBean maxwellLogBean, Context ctx, Collector<DynamicKeyedBean> collector) throws Exception {

        broadcastState = ctx.getBroadcastState(StateDescContainer.ruleStateDesc);
        //规则平台上操作的规则数据
        List<RuleTableRecord> ruleTableRecords = maxwellLogBean.getData();
        //规则平台上的操作类型
        String operationType = maxwellLogBean.getType();
        //根据不同的操作类型去操作state中的规则数据
        for (RuleTableRecord ruleTableRecord : ruleTableRecords) {
            String rule_condition_json = ruleTableRecord.getRule_condition_json();
            String rule_controller_drl = ruleTableRecord.getRule_controller_drl();
            String rule_name = ruleTableRecord.getRule_name();
            MarketingRule marketingRule = JSON.parseObject(rule_condition_json, MarketingRule.class);
            String keyByFields = marketingRule.getKeyByFields();
            String ruleName = marketingRule.getRuleId();
            String rule_status = ruleTableRecord.getRule_status();
            List<RuleStateBean> ruleStateBeanList = broadcastState.get(keyByFields);;
            //如果是插入操作，或者是设置(1的更新操作)<ip,list(ruleStateBean1,ruleStateBean2,ruleStateBean3>
            if ("insert".equals(operationType) || ("update".equals(operationType) && "1".equals(rule_status))) {
                if (ruleStateBeanList == null) {
                    ruleStateBeanList = new ArrayList<>();
                }
                //在该list放入一个新的规则bean对象
                KieSession kieSession = new KieHelper().addContent(rule_controller_drl, ResourceType.DRL).build().newKieSession();
                RuleStateBean ruleStateBean = new RuleStateBean(marketingRule, kieSession);
                ruleStateBeanList.add(ruleStateBean);
                broadcastState.put(keyByFields,ruleStateBeanList);

            } else if ("delete".equals(operationType) || ("update".equals(operationType) && "0".equals(rule_status))) {
                Iterator<RuleStateBean> iterator = ruleStateBeanList.iterator();
                while(iterator.hasNext()){
                    RuleStateBean stateBean = iterator.next();
                    if (stateBean.getMarketingRule().getRuleId().equals(rule_name)) iterator.remove();
                }
                //如果keyby对应的ruleList已经被删除，则直接从广播state中将这个keybyFields条目都删除
                if (ruleStateBeanList.size()<1) {
                    broadcastState.remove(keyByFields);
                }
            }
        }
    }
}
